package cn.yjh.server.net;

import java.io.IOException;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.ServerSocketChannel;
import java.nio.channels.SocketChannel;
import java.util.HashSet;
import java.util.Iterator;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.LockSupport;

import cn.yjh.server.Handler;
import cn.yjh.support.ServiceHandler;
import cn.yjh.utils.ClassUtils;
import cn.yjh.server.support.NamedThreadFactory;
import cn.yjh.annotation.RPCService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;


/**
 * @author yW0041229
 * @since 2021/12/13 16:34
 */
public class Acceptor implements Handler {

    Logger logger = LoggerFactory.getLogger(Acceptor.class);

    private final ServerSocketChannel serverSocketChannel;

    // 获取 CPU 核心数
    private final int coreNum = 8;//Runtime.getRuntime().availableProcessors();

    // 创建 selector 给 SlaveReactor 使用，个数为 CPU 核心数
    private final Selector[] selectors = new Selector[coreNum];

    // 轮询使用 SlaveReactor 的下标索引
    private int next = 0;

    // SlaveReactor 组
    private SlaveReactor[] reactors = new SlaveReactor[coreNum];

    // SlaveReactor 的处理线程组
    private Thread[] threads = new Thread[coreNum];

    // 处理任务的线程池
    private final ThreadPoolExecutor workers = new ThreadPoolExecutor(50, 200, 10,
            TimeUnit.SECONDS, new LinkedBlockingQueue<>(200), new NamedThreadFactory("rpc-nio"),new ThreadPoolExecutor.DiscardPolicy());

    protected ServiceHandler serviceHandler = null;

    Acceptor(ServerSocketChannel serverSocketChannel, String scanPackage) throws IOException {

        serviceHandler = new ServiceHandler(scanPackage);

        this.serverSocketChannel = serverSocketChannel;

        logger.info("init slave reactor, the slave reactor num is  [ {} ]",coreNum);

        // 初始化从 Reactor
        for (int i = 0; i < coreNum; i++) {

            selectors[i] = Selector.open();

            // 初始化sub reactor
            reactors[i] = new SlaveReactor(selectors[i], i);

            // 开启一个新的线程来处理 SlaveReactor
            threads[i] = new Thread(reactors[i]);

            // 设置 SlaveReactor 名称
            threads[i].setName("SALVE-"+i);

            //启动 SlaveReactor
            threads[i].start();
        }
    }

    @Override
    public void process() {
        SocketChannel socketChannel;
        try {
            // 获取客户端的连接
            socketChannel = serverSocketChannel.accept();

            if (socketChannel != null) {

                logger.debug("connected ===> [local={} remote={}]",socketChannel.getLocalAddress(),socketChannel.getRemoteAddress());

                // 设置当前连接为非阻塞
                socketChannel.configureBlocking(false);

                reactors[next].setStatus("run");

                LockSupport.unpark(threads[next]);

                // 设置 disableSelect 为 true，禁止执行 selector.select() 方法
                reactors[next].setDisableSelect(true);

                // 使阻塞住的 select() 方法立即返回
                selectors[next].wakeup();

                // 将 socketChannel 注册到 selector 上，并且关注读就绪事件
                SelectionKey selectionKey = socketChannel.register(selectors[next], SelectionKey.OP_READ);

                // 设置 disableSelect 为 false，让 SlaveReactor 正常执行 selector.select() 方法
                reactors[next].setDisableSelect(false);

                // 创建处理 sockechannel 的 HandlerManager
                DefaultHandler asyncHandler = new DefaultHandler(reactors[next],threads[next],socketChannel, selectionKey, next, workers,serviceHandler);

                // 绑定Handler
                selectionKey.attach(asyncHandler);

                // 当接收到一个连接后，采用轮询的方式，决定让那个 SlaveReactor 去处理
                if (++next == selectors.length) {

                    //越界后重新分配
                    next = 0;
                }


            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

}
